/*  
  -- The Lime II Project -- 

  A tuplespaces-based middleware for coordinating agents and mobile hosts.
  Copyright (C) 2005.
  Gian Pietro Picco, Amy L. Murphy, Lorenzo Bellini.

  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation; either
  version 2.1 of the License, or (at your option) any later version.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  Lesser General Public License for more details.

  You should have received this copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 
*/

package lime2;
import lights.interfaces.*;
import java.util.*;

import sagroup.dracots.SignalTuple;

/**
 * Provides the core support for a reactive tuplespace, on a sole strong reactions basis. 
 * <br>
 * <br>
 * <b><i>Basics of Strong Reactions</i></b>
 * <br>
 * <br>
 * Strong reactions are installed by the application programmers or by the Lime system itself (in this case we call 
 * them strong "system reactions"). They can be triggered only when a new tuple is inserted or when the reaction is
 * installed. Strong reactions guarantee atomicity of being enabled / firing, but they can be registered only for
 * the local (= this) host (see the Lime papers; for details about atomicity of execution, see <i>LimeTSMgr</i>). 
 * <br>
 * <br>
 * <b><i>The Reactor</i></b>
 * <br>
 * <br>
 * The methods of reactor are invoked only by the ReactionMgr (= system thread); the firing of a strong reaction, that
 * is to say calling "reactsTo" of the corresponding ReactionListener, is performed by the reactor itself.
 * Reactor stores
 * <P>
 * &#9679 the strong reactions in a linked list of LinkedRegisteredReaction instances
 * <br>
 * &#9679 the tuples in a private ITupleSpace proxy + Vector "workingTuples" + Vector "newTuples"
 * </P>
 * When the reactor is running (= the reactive program is running), it goes through the linked list (from the firstRR
 * to the lastRR) looking for enabled reactions and it makes them fire. We use a linked list because we want to 
 * <b>optimize seek speed</b>, since - as we will see - the reactor goes through the list <i>many</i> times.
 * <br>
 * Reactor makes also use of an <b>aging optimization</b>: each reaction searches the tuplespace only one time time in
 * its life.. since the tuplespace is usually BIG). The tuples that are searched each time by every reaction of the 
 * linked list are a (relatively) small set: the "working tuples" (stored in the private Vector "workingTuples"). 
 * After that the "working tuples" have been looked at by each reaction, they become old and so they go into the 
 * tuplespace.
 * <br>
 * If a reaction performs the out of a tuple, this tuple goes into the "new tuples" (stored in the private Vector 
 * "newTuples"); "new tuples" are tuples written during a reaction (-> the reactor is running); let's call "x" the 
 * reaction that performs out of tuple "y". All the reactions that precede "x" in the linked list have not looked at 
 * the "y" tuple. So "new tuples" at the end of the reactive program become "working tuples": in this way they will be
 * examined again by the reactions (-> in particular by the reactions the preceded "x" in the linked list). If a 
 * reaction fires, the reactor will run again the reactive program, restarting from the beginning of the linked list, 
 * until quiescence is achieved.
 * <br>
 * <br>
 * <b><i>Three Notes</i></b>
 * <br>
 * <br>
 * <i>Modes</i>
 * <br>
 * Once reactions are automatically uninstalled after that they are fired; once-per-tuple reactions store the TupleIDs
 * of the tuples they react to (see method "alreadyReactedTo" of class <i>RegisteredReaction</i>) in order not to react
 * twice or more on the same tuple. 
 * <br>
 * Please note that Lime doesn't allow the "out" of a tuple with a given TupleID: this could interfere with reactions,
 * i.e., re-inserted tuples wouldn't re-trigger once-per-tuple reactions.
 * <br>
 * <br>
 * <i>Deep Copying</i>
 * <br> 
 * Please note that the ITuples returned by the methods of the reactor are deep copies of the ones it stores (if the 
 * tuple is returned from "working tuples" or "new tuples" then "LimeTupleSpace.deepCopy" is invoked; if it returned 
 * from the internal ITupleSpace object then "ITupleSpace.rd/rdp/rdg/.." <u>should</u> enforce the semantics of deep 
 * copying - see class<i>TupleSpaceProxy</i> for an exhaustive discussion). 
 * <br>
 * Also the <u>triggering tuples</u> are deep copied before instantiating the ReactionEvents that will be passed to 
 * the reaction listeners of the strong reactions.
 * <br>
 * Thus if you modify the tuples returned by the reactor you don't modify the corresponding tuples still in the 
 * reactor. The non-trivial consequence is that you can manipulate the contents of a LimeTupleSpace only using the in,
 * rd, out, etc.. methods and not, i.e., changing the fields of an ITuple returned by a "rd" on the LimeTupleSpace. 
 * <br>
 * <br>
 * <i>Calibrating Group Write Operation</i>
 * <br>
 * When the "workingTuples" become old, the reactor writes them in the tuplespace. There are two ways to do this: 
 * performing multiple "out" operations, as it was in Lime 1.11 or performing a single "outg" operation.
 * <br> 
 * The default policy is to force a single outg operation when the number of tuples to write is greater than one. But 
 * the threshold can be changed (see <i>LimeConstants</i>). This can be useful to calibrate overhead of the underlying
 * tuplespace engine, although we expect the "outg" method to be more and more efficient, as the number of tuples to
 * write increases.
 * 
 * @author Gian Pietro Picco
 * @author Amy Murphy
 * @author Lorenzo Bellini
 */

class Reactor 
{
 // hashtable of strong reactions: it is managed by the LimeTSMgr, but since Reactor automatically removes strong once
 // reactions after firing, Reactor needs to update it too
 private Hashtable registry;	
	
 // the tuplespace layer 
 private ITupleSpace ts;
 
 // keeps track wheter the reactor is running now 
 private boolean running = false; 
  
 // a pointer to the first registered reaction within the linked list
 private LinkedRegisteredReaction firstRR = null;

 // a pointer to the last registered reaction within the linked list
 private LinkedRegisteredReaction lastRR = null;

 // a pointer to the current registered reaction within the linked list
 private LinkedRegisteredReaction currentRR = null;

 // holds (1) tuples that are being passed to the reactor upon triggering its computation 
 //       (2) tuples that have been dropped by an out performed within a weak reaction 
 private Vector workingTuples = new Vector(100, 100);

 // holds the tuples that have been dropped by an out performed within a strong reaction
 private Vector newTuples = new Vector(100, 100); 

 //	auxiliary vector
 private Vector matchingTuples = new Vector(100, 100);
 
 // the tuples that eliminate all the "SignalTuple"
 private Vector shareTuples = new Vector(100, 100);

 /**
  * Constructor method.
  * 
  * @param ts the (proxied) tuplespace
  * @param registry the hashtable of strong reactions: it is initialized and managed by the LimeTSMgr, but since 
  *        Reactor automatically removes strong once reactions after firing, Reactor needs to update it too
  * @throws TupleSpaceEngineException
  */
 Reactor(ITupleSpace ts, Hashtable registry) 
 {
  // sets the tuplespace	
  this.ts = ts;	
  // sets the hashtable of strong reactions
  this.registry = registry;	
 }

 /**
  * Merges (FIFO) the given registered reactions in the reactor. Please note that Reactor optimizes seek speed using
  * a linked list: class LinkedRegisteredReaction is a "list-enabled" subtype of RegisteredReaction.  
  *
  * @param lrr array of LinkedRegisteredReactions
  */
 void put(LinkedRegisteredReaction[] lrr) 
 {
  for (int i = 0; i < lrr.length; i++) 
  {
   // initializes marker	
   lrr[i].isNew = true;
       
   // manages the linked list of reactions 
   if (firstRR == null) 
   {
	lrr[i].prev = null;
	firstRR = lrr[i];
	lastRR = lrr[i];
	lrr[i].next = null;
   } 
   else 
   {
	lastRR.next = lrr[i];
	lrr[i].prev = lastRR;
	lastRR = lrr[i];
	lastRR.next = null;
   }
  }
 }

 /**
  * Tries to remove an array of reactions from the linked list of registered reactions. As a side effect, this 
  * method may also update the next pointer of the current execution of the reactive program. 
  * If a reaction has been successfully deregistered, the corresponding boolean - within the returned array of 
  * booleans - is set to true, otherwise it is set to false. 
  * Please note that this method needs to update the hashtable of strong reactions.
  *
  * @param rids array of ReactionIDs of the reactions to remove
  * @return boolean[], where <code>boolean[i]</code> is <code>true</code> if reaction <code>i</code> was successfully
  *         removed, <code>false</code> otherwise
  */
 boolean[] remove(ReactionID[] rids) 
 {
  boolean[] ret = new boolean[rids.length];
 	
  for (int i = 0; i < rids.length; i++) 
  {
   // the only line of Reactor that bounds it to the LimeTSMgr (LimeTSMgr is bounded to Reactor) 	
   Object found = registry.remove(rids[i]);
   
   if (found == null) ret[i] = false;
   else
   {	
    ret[i] = true;
    LinkedRegisteredReaction xx = (LinkedRegisteredReaction) found;  	
       
    // removing last element
    if (xx.next == null) lastRR = xx.prev; 
    else xx.next.prev = xx.prev;

    // removing first element 
    if (xx.prev == null) firstRR = xx.next; 
    else xx.prev.next = xx.next;

    // removing element next to current element 
    if (currentRR != null && currentRR.next == xx) currentRR.next = xx.next; 
 
    // it is not necessary to worry about removing the current element: see the "runReactiveProgram" method;
    // the current reaction is not garbage-collected if it is removed from the vector of reactions because 
    // there are still references to it (r, currentRR; r = currentRR).
    // Only when the reactive programs moves to the next reaction, it will be garbage-collected 
   }
  }
 
  return ret;  
 } 

 /**
  * Returns whether the reactor is running now. If the reactor is running then:
  * <P>
  * (1) it could be executing ("within" the listener of) a strong reaction
  * <br>
  * (2) it could be just performing the cycle-scan of reactions / tuples 
  * </P>
  *
  * @return <code>boolean</code>
  */
 boolean isRunning() 
 {
  return running;
 }
 
 /**
  * Appends an array of tuples to vector "workingTuples".
  * 
  * @param t array of tuples to write into vector "workingTuples".
  */
 void writeWorkingTuples(ITuple[] t) 
 {
  for (int i = 0; i < t.length; i++) workingTuples.addElement(t[i]); 
 }
  
 /**
  * Appends an array of tuples to vector "newTuples".
  * 
  * @param t array of tuples to write into vector "newTuples".
  */
 void writeNewTuples(ITuple[] t) 
 {
  for (int i = 0; i < t.length; i++) 
  {
	  newTuples.addElement(t[i]); 

  }
  
 }

 /**
  * Reads a tuple (returns a deep copy, see <i>TupleSpaceProxy</i>).
  * 
  * @param t template
  * @return tuple found
  */
 ITuple doRead(ITuple t) 
 {
  ITuple result = null;
  // searches ts (1), workingTuples (2), newTuples (3) - in this strict order
  try { result = ts.rdp(t); } catch (TupleSpaceException tse) { throw new LimeRuntimeException("tuplespace engine error", tse); }
  if (result == null) result = findMatch(workingTuples, t);
  if (result == null) result = findMatch(newTuples, t);
  return result;
 }

 /**
  * Reads a group of tuples (returns deep copies, see <i>TupleSpaceProxy</i>). 
  * 
  * @param t template
  * @return tuples found
  */
 ITuple[] doReadg(ITuple t) 
 {
  ITuple[] result = null;
  ITuple[] ot = null;
  ITuple[] wt = null;
  ITuple[] nt = null;
  int length = 0;
  int otLength = 0;
  int wtLength = 0;
  int ntLength = 0;
   
  // searches ts, workingTuples, newTuples
  try { ot = ts.rdg(t); } catch (TupleSpaceException tse) { throw new LimeRuntimeException("tuplespace engine error", tse); }
  wt = findAllMatches(workingTuples, t);
  nt = findAllMatches(newTuples, t);
  
  if (ot != null) otLength = ot.length;
  if (wt != null) wtLength = wt.length;
  if (nt != null) ntLength = nt.length;
  length = otLength + wtLength + ntLength;
  
  if (length == 0) result = null;
  else 
  {
   result = new ITuple[length];
   if (ot != null) System.arraycopy(ot, 0, result, 0, otLength);
   if (wt != null) System.arraycopy(wt, 0, result, otLength, wtLength);
   if (nt != null) System.arraycopy(nt, 0, result, otLength + wtLength, ntLength);
  }
  return result;
 }

 /**
  * Takes a tuple (= returns a deep copy - see <i>TupleSpaceProxy</i> - and removes the original tuple).
  * 
  * @param t template
  * @return tuple found
  */
 ITuple doTake(ITuple t) 
 {
  ITuple result = null;
  
  // searches (and removes from) ts (1), workingTuples (2), newTuples (3)- in this strict order
  try { result = ts.inp(t); } catch (TupleSpaceException tse) { throw new LimeRuntimeException("tuplespace engine error", tse); }
  if (result == null) 
  { 
   result = findMatch(workingTuples, t); 
   removeMatch(workingTuples, new ITuple[] {result});
  }
  if (result == null) 
  { 
   result = findMatch(newTuples, t);
   removeMatch(newTuples, new ITuple[] {result});
  }
  // remove the tuple also from the matchingTuples
  removeMatch(matchingTuples, new ITuple[] {result});

  return result;
 }

 /**
  * Takes a group of tuples (= returns deep copies - see <i>TupleSpaceProxy</i> - and removes the original tuples).
  * 
  * @param t template
  * @return tuples found 
  */
 ITuple[] doTakeg(ITuple t) 
 {
  ITuple[] result = null;
  ITuple[] ot = null;
  ITuple[] wt = null;
  ITuple[] nt = null;
  int length = 0;
  int otLength = 0;
  int wtLength = 0;
  int ntLength = 0;
 
  // searches (and removes from) ts, workingTuples, newTuples
  try { ot = ts.ing(t); } catch (TupleSpaceException tse) { throw new LimeRuntimeException("tuplespace engine error", tse); }
  wt = findAllMatches(workingTuples, t);
  nt = findAllMatches(newTuples, t);
  removeMatch(workingTuples, wt);
  removeMatch(newTuples, nt);
 
  if (ot != null) otLength = ot.length;
  if (wt != null) wtLength = wt.length;
  if (nt != null) ntLength = nt.length;
  length = otLength + wtLength + ntLength;
  
  if (length == 0) result = null;
  else 
  {
   result = new ITuple[length];
   if (ot != null) System.arraycopy(ot, 0, result, 0, otLength);
   if (wt != null) System.arraycopy(wt, 0, result, otLength, wtLength);
   if (nt != null) System.arraycopy(nt, 0, result, otLength + wtLength, ntLength);
   //remove the tuples also from the matchingTuples 
   removeMatch(matchingTuples, result);
  }
   
  return result;
 }

 /**
  * Runs the reactive program ("runs the reactor").
  *
  * @param wt tuples to append to the workingTuples (they can be passed also using the "writeWorkingTuples" method)
  */
 void runReactiveProgram(ITuple[] wt) 
 {
  running = true;	
  boolean enabled = true;
  boolean reacted = false;

  // if some tuples were sent (not null)
  if (wt != null) for (int j = 0; j < wt.length; j++) workingTuples.addElement(wt[j]);
   
  // keep running as long as there are reactions enabled
  while(enabled) 
  {
   enabled = false;
   currentRR = firstRR;
   LinkedRegisteredReaction r = null;
   ITuple template = null;
   ITuple t = null;
   ITuple[] ta = null;

   while (currentRR != null) 
   {
	// process the current registered reaction
	r = currentRR;
	template = r.getTemplate();
	t = null;
	reacted = false;
	// resets the bag of matching tuples
	matchingTuples.clear();

	// we search the tuplespace (which is supposed to be quite big) only if the reactor never 
	// processed this reaction before (OPTIMIZATION)
	if (r.isNew) 
	{
	 r.isNew = false;

	 switch(r.getMode()) 
	 {
	  case Reaction.ONCE:
	       try { t = ts.rdp(template); } catch (TupleSpaceException tse) { throw new LimeRuntimeException("tuplespace engine error", tse); }
  	       if (t != null) 
	       {
	        reacted = true;
	        reactOnce(t, r);
	       }
	       break;
      case Reaction.ONCEPERTUPLE:
	       try { ta = ts.rdg(template);	} catch (TupleSpaceException tse) { throw new LimeRuntimeException("tuplespace engine error", tse); }
		   if (ta != null) for (int i = 0; i < ta.length; i++) matchingTuples.addElement(ta[i]);
	 } // end switch
    }  // end if (r.isNew)

	switch(r.getMode()) 
	{
	 case Reaction.ONCE:
		  // we need to look both at workingTuples and newTuples to search for matching tuples we can react to
		  if (!reacted) reacted = reactOnce(workingTuples, template, r);
		  if (!reacted) reacted = reactOnce(newTuples, template, r);
		  break;
     case Reaction.ONCEPERTUPLE:
		  // we fire on matchingTuples (.. "reactSelectely" re-checks but it is unnecessary) 
		  // we search workingTuples, and newTuples for other matching tuples we can fire on 
		  reacted = reactSelectively(matchingTuples, template, r);
		  boolean bool1 = reactSelectively(workingTuples, template, r);
		  boolean bool2 = reactSelectively(newTuples, template, r);
		  reacted = reacted || bool1 || bool2;
		  break;
	} // end switch

    enabled = enabled || reacted;
	currentRR = currentRR.next;
   } // end while (currentRR != null)

   // move the tuples in workingTuples to "oldTuples", that is to say, to the tuplespace
   try 
   {
   	int l = workingTuples.size();
   	//filter out all the Signal Tuple. That's any signal tuple will never really written into tuple space
   	for(int i = 0; i < l; i++) {
   		ITuple xt = (ITuple) workingTuples.elementAt(i);
   		if(xt.length() > 5){
	   		IField xf = xt.get(5); // acquire the clzName, this assumes the accepted tuple is flat representation of "DracoTuple"
	   		if(!xf.isFormal() && xf.getType().equals(String.class)) {
	   			String clzName = (String)((IValuedField)xf).getValue();
	   			try {
	   				// if it is a sub class of SignalTuple, just ignore it
					if (SignalTuple.class.isAssignableFrom(Class.forName(clzName))) {
						continue;
					}
				} catch (ClassNotFoundException e) {	
					continue;
				}
	   		}
   		}
   		shareTuples.add(xt);
   	}
   	Vector tmp = workingTuples;
   	workingTuples = shareTuples;
   	tmp.clear();
   	shareTuples = tmp;
   	
   	// move the share tuples into the tuple space
   	l = workingTuples.size();
   	if (l > 0)
   	   {	
   		// if the number of tuples to out exceeds FORCE_GROUP_OPERATION then Reactor switches from many "out" 
   		// operations to a single "outg" operation. This can be useful to calibrate write overhead of the underlying 
   		// tuplespace engine
   	    if (l <= LimeConstants.FORCE_GROUP_OPERATION)
   	    	for (int j = 0; j < l; j++) 
   	    	ts.out((ITuple) workingTuples.elementAt(j)); 
   	    else 
   	    { 
   	      ITuple[] tempaux = new ITuple[l]; 
   	      workingTuples.copyInto(tempaux);
   	     ts.outg(tempaux); }	
   	   } 
   } 
   catch (TupleSpaceException tse) { throw new LimeRuntimeException("tuplespace engine error", tse); }

   // move the tuples in newTuples to workingTuples and reset the newTuples
   Vector tmp = workingTuples; 
   workingTuples = newTuples;
   tmp.clear();  
   newTuples = tmp; 
   matchingTuples.clear();
  }
 
  running = false;
 }

 // reacts to tuple "t" invoking the listener of the registered reaction "r": please note that the passed 
 // ReactionEvent wraps a deep copy of the triggering tuple
 private void react(ITuple t, RegisteredReaction r) 
 {
  ReactionEvent e = new ReactionEvent(LimeTupleSpace.deepCopy(t));
  r.getListener().reactsToOut(e);
 }

 // fires the registered reaction using the specified tuple. Then removes the registered reaction from this reactor
 private void reactOnce(ITuple t, RegisteredReaction r) 
 {
  remove(new ReactionID[] {r.getReactionID()});	
  react(t, r);
 }

 // tries to react once on those tuples in the multiset that match the tuple. Returns true (and stops) as soon as 
 // the registered reaction fires
 private boolean reactOnce(Vector tuples, ITuple template, RegisteredReaction r) 
 {
  boolean result = false;
  if (tuples.size() != 0) 
  {
   ITuple tj = null;
   for (int j = 0; j < tuples.size() && !result; j++) 
   {
    tj = (ITuple) tuples.elementAt(j);
    if (TupleSpaceProxy.matchTuples(template, tj)) 
    {
     reactOnce(tj, r);
	 result = true;
	}
   }
  }
  return result;
 }

 // tries to react once-per-tuple on those tuples in the multiset that match the tuple. The information on whether 
 // the tuple is already reacted to or not is contained in the registered reaction. Returns true if at least a 
 // reaction took place
 private boolean reactSelectively(Vector tuples, ITuple template, RegisteredReaction r) 
 {
  boolean result = false;
  if (tuples.size() != 0) 
  {
   ITuple tj = null;
   for (int j=0; j < tuples.size(); j++) 
   {
	tj = (ITuple) tuples.elementAt(j);
	if (TupleSpaceProxy.matchTuples(template, tj) && !r.alreadyReactedTo(LimeWrapper.getID(tj))) 
	{
     result = true;
	 react(tj, r);
	}
   }
  }
  return result;
 }

 // returns the (deep copy of) first ITuple of Vector "tuples" matching the given template (or null if no match is 
 // found)
 private static ITuple findMatch(Vector tuples, ITuple template) 
 {
  ITuple result = null;
  boolean found = false;
  ITuple tj = null;
  int i = 0;
  while (i < tuples.size() && !found) 
  {
   tj = (ITuple) tuples.elementAt(i++);
   found = TupleSpaceProxy.matchTuples(template, tj);
  }
  if (found) result = tj;
  return LimeTupleSpace.deepCopy(result);
 }

 // return an array (of deep copies) of the ITuples of Vector "tuples" matching the given template (or an array of 
 // length zero if no match is found)
 private static ITuple[] findAllMatches(Vector tuples, ITuple template) 
 {
  ITuple[] result = null;
  Vector v = new Vector(tuples.size(),0);
  ITuple tj = null;
  int i = 0;
  while (i < tuples.size()) 
  {
   tj = (ITuple) tuples.elementAt(i++);
   if (TupleSpaceProxy.matchTuples(template, tj)) v.addElement(LimeTupleSpace.deepCopy(tj));
  }
  result = new ITuple[v.size()];
  v.copyInto(result);
  return result;
 }

 // removes an array of ITuples ("t") from a Vector of ITuples ("tuples")
 private static void removeMatch(Vector tuples, ITuple[] t) 
 {
  for (int j = 0; j < t.length; j++) 
  {
   if (t[j] != null) 
   {
    TupleID id = LimeWrapper.getID(t[j]);
	boolean found = false;
    ITuple aTuple = null;
    TupleID aID = null;
	int i = 0;
    for (; i < tuples.size() && !found; i++) 
    {
 	 aTuple = (ITuple) tuples.elementAt(i); 
	 aID = LimeWrapper.getID(aTuple);
	 found = aID.equals(id);
    }
    if (found) tuples.remove(--i);
   }
  }
 }
 
}
